package myapps4;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * consumer 同步手动提交位移 ，减少重复消费的问题出现，但仍然会出现重复消费
 */
public class SimpleConsumer {
	private static final String GROUPID = "groupA";
	private static final String topic = "KAFKA_TEST";

	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.0.150:9092");
		props.put("group.id", GROUPID);
//		props.put("enable.auto.commit", "true");
//		props.put("auto.commit.interval.ms", "1000");
		//设置手动提交
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
		props.put("session.timeout.ms", "30000");
		props.put("auto.offset.reset", "earliest");
		props.put("key.deserializer", StringDeserializer.class.getName());
		props.put("value.deserializer", StringDeserializer.class.getName());
		KafkaConsumer consumer = new KafkaConsumer<String, String>(props);
//		consumer.subscribe(Arrays.asList(topic));
		//指定分区（可选）
		TopicPartition tp = new TopicPartition(topic, 0);
		consumer.assign(Arrays.asList(tp));
		long offset = -1L;
		while(true) {
			ConsumerRecords<String, String> msgList = consumer.poll(10000);
			if (msgList != null && !msgList.isEmpty()) {
				for (ConsumerRecord<String, String> record : msgList) {
					System.out.println("=======receive: key = " + record.key() + ", value = " + record.value() + " offset===" + record.offset());
				}
				List<ConsumerRecord<String, String>> records = msgList.records(tp);
				int size = records.size();
				System.out.println(size);
				offset = records.get(size - 1).offset();
				consumer.commitSync();
			}
			break;
		}
		System.out.println("last offset:"+offset);
		OffsetAndMetadata committed = consumer.committed(tp);
		long nextLocation = committed.offset();
		System.out.println("next offset:"+nextLocation);
	}
}
